package com.atguigu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.app.func.AsyncJoinFunction;
import com.atguigu.bean.TradeCourseOrderWindow;
import com.atguigu.util.ClickHouseUtil;
import com.atguigu.util.DateFormatUtil;
import com.atguigu.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.async.AsyncFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;
import scala.tools.nsc.doc.model.Val;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

public class DwsTradeCourseOrderWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //TODO 2 设置状态后端
        /*
        env.enableCheckpointing(5*60*1000L, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(3*60*1000L);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        System.setProperty("HADOOP_USER_NAME","atxiaomeng");
        */

        //TODO 3读取kafka dwd_course_order_detail主题的数据
        String topicName = "dwd_course_order_detail";
        String groupId = "dws_trade_course_order_window";
        DataStreamSource<String> dtStream = env.addSource(KafkaUtil.getFlinkKafkaConsumer(topicName, groupId));
//        dtStream.print();


        //TODO 4过滤转化结构 javabean
        SingleOutputStreamOperator<TradeCourseOrderWindow> orderBeanStream = dtStream.flatMap(new FlatMapFunction<String, TradeCourseOrderWindow>() {
            @Override
            public void flatMap(String value, Collector<TradeCourseOrderWindow> out) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                out.collect(TradeCourseOrderWindow.builder()
                        .course_id(jsonObject.getString("course_id"))
                        .orderCount(1L)
                        .final_amount(jsonObject.getDouble("final_amount"))
                        .ts(jsonObject.getLong("ts")*1000L)
                        .build());
            }
        });

        //TODO 5 设置水位线
        SingleOutputStreamOperator<TradeCourseOrderWindow> withWaterMarkStream = orderBeanStream.assignTimestampsAndWatermarks(WatermarkStrategy.<TradeCourseOrderWindow>forBoundedOutOfOrderness(Duration.ofSeconds(2L)).withTimestampAssigner(new SerializableTimestampAssigner<TradeCourseOrderWindow>() {
            @Override
            public long extractTimestamp(TradeCourseOrderWindow value, long timeStamp) {
                return value.getTs();
            }
        }));

        //TODO 6按照课程id key by 开窗聚合
        SingleOutputStreamOperator<TradeCourseOrderWindow> reduceStream = withWaterMarkStream.keyBy(new KeySelector<TradeCourseOrderWindow, String>() {
            @Override
            public String getKey(TradeCourseOrderWindow value) throws Exception {
                return value.getCourse_id();
            }
        }).window(TumblingEventTimeWindows.of(Time.seconds(10L)))
                .reduce(new ReduceFunction<TradeCourseOrderWindow>() {
                    @Override
                    public TradeCourseOrderWindow reduce(TradeCourseOrderWindow value1, TradeCourseOrderWindow value2) throws Exception {
                        value1.setFinal_amount(value1.getFinal_amount() + value2.getFinal_amount());
                        value1.setOrderCount(value1.getOrderCount() + value2.getOrderCount());
                        return value1;
                    }
                }, new WindowFunction<TradeCourseOrderWindow, TradeCourseOrderWindow, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow timeWindow, Iterable<TradeCourseOrderWindow> value, Collector<TradeCourseOrderWindow> out) throws Exception {
                        TradeCourseOrderWindow next = value.iterator().next();
                        next.setStt(DateFormatUtil.toYmdHms(timeWindow.getStart()));
                        next.setEdt(DateFormatUtil.toYmdHms(timeWindow.getEnd()));
                        next.setTs(System.currentTimeMillis());
                        out.collect(next);
                    }
                });

        reduceStream.print("reduceStream");
        //TODO 7关联维度表
        //7.1 关联课程表
            SingleOutputStreamOperator<TradeCourseOrderWindow> coursubjectIdDtStream = AsyncDataStream
                    .unorderedWait(reduceStream, new AsyncJoinFunction<TradeCourseOrderWindow>("DIM_COURSE_INFO") {
            @Override
            public String getKey(TradeCourseOrderWindow input) {
                return input.getCourse_id();
            }

            @Override
            public void join(TradeCourseOrderWindow input, JSONObject dimSkuInfo) {
                input.setSubject_id(dimSkuInfo.getString("subjectId"));

            }
        }, 60L, TimeUnit.SECONDS);
        coursubjectIdDtStream.print("coursubjectIdDtStream");

        //7.2 关联科目表
        SingleOutputStreamOperator<TradeCourseOrderWindow> subjectNameDtStream = AsyncDataStream.unorderedWait(coursubjectIdDtStream, new AsyncJoinFunction<TradeCourseOrderWindow>("DIM_BASE_SUBJECT_INFO") {
            @Override
            public String getKey(TradeCourseOrderWindow input) {
                return input.getSubject_id();
            }

            @Override
            public void join(TradeCourseOrderWindow input, JSONObject dimSkuInfo) {
                input.setSubject_name(dimSkuInfo.getString("subjectName"));
                input.setCategory_id(dimSkuInfo.getString("categoryId"));
            }
        }, 60, TimeUnit.SECONDS);
        subjectNameDtStream.print("subjectNameDtStream");
        //7.3 关联分类表
        SingleOutputStreamOperator<TradeCourseOrderWindow> catSubStream = AsyncDataStream.unorderedWait(subjectNameDtStream, new AsyncJoinFunction<TradeCourseOrderWindow>("DIM_BASE_CATEGORY_INFO") {
            @Override
            public String getKey(TradeCourseOrderWindow input) {
                return input.getCategory_id();
            }

            @Override
            public void join(TradeCourseOrderWindow input, JSONObject dimSkuInfo) {
                input.setCategory_name(dimSkuInfo.getString("categoryName"));
            }
        }, 60, TimeUnit.SECONDS);
        catSubStream.print("catSubStream");

        //TODO 写出到clickhouse
        catSubStream.addSink(ClickHouseUtil.getClickHouseSink("insert into dws_trade_course_order_window values(?,?,?,?,?,?,?,?,?,?)"));

        //TODO 执行任务
        env.execute(groupId);

    }
}
